package com.asura.ops.sync.server.service.impl;

import cn.hutool.json.JSONConfig;
import cn.hutool.json.JSONUtil;
import com.asura.ops.sync.server.model.entity.CfgDbEntity;
import com.asura.ops.sync.server.model.entity.CfgMqEntity;
import com.asura.ops.sync.server.model.entity.ChangeInfoEntity;
import com.asura.ops.sync.server.model.entity.ChangeInfoMqEntity;
import com.asura.ops.sync.server.model.entity.req.RepushByClientCodeReq;
import com.asura.ops.sync.server.model.entity.req.RepushReq;
import com.asura.ops.sync.server.mq.DynamicMQ;
import com.asura.ops.sync.server.repush.DataFilterInfo;
import com.asura.ops.sync.server.repush.DataManage;
import com.asura.ops.sync.server.service.CfgMqService;
import com.asura.ops.sync.server.service.ChangeInfoMqService;
import com.asura.ops.sync.server.service.ChangeInfoService;
import com.asura.ops.sync.server.service.ManageService;
import com.asura.ops.sync.server.sync.consume.ChangePayload;
import com.asura.ops.sync.server.sync.consume.ChangeSource;
import com.asura.ops.sync.server.sync.enums.ChangeTypeEnum;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

/**
 * @author: huyuntao(Mars)
 * @date: 2022/8/10
 * @description: 类的描述
 */
@Service
@Slf4j
public class ManageServiceImpl implements ManageService {

    @Autowired
    private DynamicMQ dynamicMQ;

    @Autowired
    private CfgMqService mqService;

    @Autowired
    private ChangeInfoService changeInfoService;

    @Autowired
    private ChangeInfoMqService changeInfoMqService;

    @Value("${sync.change.clean.day:4}")
    private Integer cleanChangeInfoDay;

    /**
     * 清理开关，通过接口可以控制
     */
    private boolean cleanSwitch = true;

    private final String DAY_FORMAT = "yyyy-MM-dd HH:mm:ss";

    private ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(1);

    private JSONConfig jsonConfig = new JSONConfig();

    @PostConstruct
    public void cleanChange() {
        jsonConfig.setIgnoreNullValue(false);

        //部署多台，可能出现同时执行，先手动执行清理，后续可以接入JOB
//        threadPool.scheduleAtFixedRate(new Runnable() {
//            @Override
//            public void run() {
//                try {
//                    String time = timeCalculate(DateTime.now().toDate(), DAY_FORMAT, Calendar.DAY_OF_MONTH, -1 * cleanChangeInfoDay);
//                    cleanChangeInfo(time);
//                } catch (ParseException e) {
//                    log.error("清理数据库记录失败：{}", e);
//                }
//            }
//        }, 0, 2, TimeUnit.HOURS);

    }

    /**
     * 时间计算
     *
     * @param date   时间
     * @param format 时间格式
     * @param type   Calendar内置常量
     * @param num    加减值
     * @return
     * @throws ParseException
     */
    public String timeCalculate(Date date, String format, int type, int num) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat(format);
//        Date d = sdf.parse(str);
        Calendar c = Calendar.getInstance();
        c.setTime(date);
        c.add(type, num);
        return sdf.format(c.getTime());
    }

    @Async
    @Override
    public void repush(RepushReq repushReq) {
        if (CollectionUtils.isEmpty(repushReq.getClients())) {
            return;
        }
        for (CfgMqEntity mqEntity : repushReq.getClients()) {
            CfgDbEntity dbEntity = mqService.queryDBForMq(mqEntity);

            DataFilterInfo dataFilterInfo = new DataFilterInfo();
            dataFilterInfo.setStartTime(repushReq.getStartTime());
            dataFilterInfo.setEndTime(repushReq.getEndTime());
            dataFilterInfo.setDbName(mqEntity.getSyncDbName());
            dataFilterInfo.setTableName(mqEntity.getSyncTableName());
            dataFilterInfo.setUpdateTimeFieldName(repushReq.getUpdateTimeFieldName());

            Integer totalRecord = DataManage.queryDataCount(dbEntity, dataFilterInfo);
            Integer pageSize = 10000;
            if (repushReq.getPageSize() != null) {
                pageSize = repushReq.getPageSize();
            }

            Integer totalNum = totalRecord;
            if (pageSize != 1) {
                totalNum = totalRecord / pageSize + (totalRecord % pageSize == 0 ? 0 : 1);
            }

            log.info("repush {}条数据", totalRecord);

            for (Integer pageNum = 1; pageNum <= totalNum; pageNum++) {
                List<HashMap<String, Object>> mapList = DataManage.queryData(dbEntity, dataFilterInfo, pageNum, pageSize);
                log.info("pageNum:{},pageSize:{},mapList:{}", pageNum, pageSize, mapList.size());

                for (HashMap<String, Object> map : mapList) {
                    ChangePayload payload = new ChangePayload();
                    payload.setChangeType(ChangeTypeEnum.REPUSH.getValue());
                    payload.setAppointDbName(repushReq.getAppointDbName());
                    payload.setAfter(map);
                    payload.setMsgId(UUID.randomUUID().toString());
                    payload.setTs_ms(System.currentTimeMillis());

                    ChangeSource source = new ChangeSource();
                    source.setDb(mqEntity.getSyncDbName());
                    source.setTable(mqEntity.getSyncTableName());
                    payload.setSource(source);

                    String suff = "_all";

                    try {
                        dynamicMQ.sendMsg(mqEntity.getExchangeName() + suff, mqEntity.getRouteKey() + "_" + mqEntity.getSyncClientCode(), JSONUtil.toJsonStr(payload, jsonConfig));
                    } catch (Exception ex) {
                        log.error("补偿发送消息失败:{}", ex);
                    }
                }
            }
        }
    }

    @Override
    public void repushByClientCode(String clientCode, String updateTimeFieldName) {
        LambdaQueryWrapper<CfgMqEntity> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(CfgMqEntity::getSyncClientCode, clientCode);

        List<CfgMqEntity> list = mqService.list(queryWrapper);

        if (CollectionUtils.isEmpty(list)) {
            return;
        }

        RepushReq repushReq = new RepushReq();
        repushReq.setClients(list);
        repushReq.setStartTime("1970-01-01 00:00:00");
        repushReq.setEndTime("2970-01-01 00:00:00");
        repushReq.setUpdateTimeFieldName(updateTimeFieldName);

        this.repush(repushReq);
    }

    @Override
    public void repushByClientCodeV2(RepushByClientCodeReq repushByClientCodeReq) {
        LambdaQueryWrapper<CfgMqEntity> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(CfgMqEntity::getSyncClientCode, repushByClientCodeReq.getClientCode());

        List<CfgMqEntity> list = mqService.list(queryWrapper);

        if (CollectionUtils.isEmpty(list)) {
            return;
        }

        RepushReq repushReq = new RepushReq();
        repushReq.setClients(list);
        repushReq.setStartTime(repushByClientCodeReq.getStartTime());
        repushReq.setEndTime(repushByClientCodeReq.getEndTime());
        if (StringUtils.isEmpty(repushByClientCodeReq.getStartTime())) {
            repushReq.setStartTime("1970-01-01 00:00:00");
        }

        if (StringUtils.isEmpty(repushByClientCodeReq.getEndTime())) {
            repushReq.setEndTime("2970-01-01 00:00:00");
        }

        if (StringUtils.isEmpty(repushByClientCodeReq.getUpdateTimeFieldName())) {
            repushReq.setUpdateTimeFieldName("updated_at");
        }
        repushReq.setAppointDbName(repushByClientCodeReq.getAppointDbName());

        this.repush(repushReq);
    }

    /**
     * 清理指定时间之前的发送记录
     *
     * @param time
     */
    public void cleanChangeInfo(String time) {
        if (!cleanSwitch) {
            log.info("清理开关未打开------------------");
            return;
        }

        log.info("开始清理发送记录");

        SimpleDateFormat sdf = new SimpleDateFormat(DAY_FORMAT);
        try {
            sdf.parse(time);
        } catch (ParseException e) {
            log.error("日期格式不正确:{}", time);
            return;
        }

        LambdaQueryWrapper<ChangeInfoEntity> deleteQuery = new LambdaQueryWrapper<>();
        deleteQuery.le(ChangeInfoEntity::getCreatedAt, time);

        changeInfoService.remove(deleteQuery);
        log.info("changeInfo-清理完成");

        LambdaQueryWrapper<ChangeInfoMqEntity> mqDeleteQuery = new LambdaQueryWrapper<>();
        mqDeleteQuery.le(ChangeInfoMqEntity::getCreatedAt, time);

        changeInfoMqService.remove(mqDeleteQuery);
        log.info("changeInfoMq-清理完成");
    }

    @Override
    public void setCleanSwitch(Boolean cleanSwitch) {
        this.cleanSwitch = cleanSwitch;
    }
}
